#ifndef __M_CCHANNEL_H__
#define __M_CCHANNEL_H__

#include"muduo/net/TcpConnection.h"
#include"muduo/proto/codec.h"
#include<google/protobuf/map.h>

#include"../mqcommon/mq_msg.pb.h"
#include"../mqcommon/helper.hpp"
#include"../mqcommon/logger.hpp"
#include"../mqcommon/mq_proto.pb.h"
#include"../mqcommon/mq_threadpool.hpp"
#include"mqc_consumer.hpp"

#include<functional>
#include<mutex>
#include <condition_variable>
#include<memory>

namespace mq{
    using ProtobufCodecPtr = std::shared_ptr<ProtobufCodec>;
    using basicConsumerResponsePtr = std::shared_ptr<basicConsumerResponse>;
    using basicCommonResponsePtr = std::shared_ptr<basicCommonResponse>;

    class ClientChannel{
        public:

            using ptr = std::shared_ptr<ClientChannel>;
            ClientChannel(const muduo::net::TcpConnectionPtr& conn,const ProtobufCodecPtr& codec)
            :_cid(Uuidhelper::BuideUuid()),
            _codecP(codec),
            _connectionP(conn){}

            std::string getCid(){
                return _cid;
            }

            bool openChannel(){
                std::string rid = Uuidhelper::BuideUuid();
                openChannelRequest req;
                req.set_rid(rid);
                req.set_cid(_cid);

                _codecP->send(_connectionP,req);
                basicCommonResponsePtr resp = waitresponse(rid);
                return resp->ok();
            }

            bool closeChannel(){
                std::string rid = Uuidhelper::BuideUuid();
                closeChannelRequest req;    
                req.set_rid(rid);
                req.set_cid(_cid);
                
                _codecP->send(_connectionP,req);
                basicCommonResponsePtr resp = waitresponse(rid);
                return resp->ok();
            }


            bool declareExchange(const std::string& exchange_name,
                ExchangeType type,
                bool durable,
                bool auto_delete,
                google::protobuf::Map<std::string,std::string>& args){
                    std::string rid = Uuidhelper::BuideUuid();
                    declareExchangeRequest req;
                    req.set_rid(rid);
                    req.set_cid(_cid);
                    req.set_exchange_name(exchange_name);
                    req.set_durable(durable);
                    req.set_auto_delete(auto_delete);
                    req.set_type(type);
                    req.mutable_args()->swap(args);

                    _codecP->send(_connectionP,req);
                    basicCommonResponsePtr resp = waitresponse(rid);
                    return resp->ok();
                }
            void deleteExchange(const std::string& exchange_name){
                    std::string rid = Uuidhelper::BuideUuid();
                    deleteExchangeRequest req;
                    req.set_rid(rid);
                    req.set_cid(_cid);
                    req.set_exchange_name(exchange_name);

                    _codecP->send(_connectionP,req);
                    waitresponse(rid);
                    return ;
            }
            bool declareQueue(const std::string& queue_name,
                bool durable,
                bool exclusive,
                bool auto_delete,
                google::protobuf::Map<std::string,std::string>& args){
                    std::string rid = Uuidhelper::BuideUuid();
                    declareQueueRequest req;
                    req.set_rid(rid);
                    req.set_cid(_cid);
                    req.set_queue_name(queue_name);
                    req.set_durable(durable);
                    req.set_auto_delete(auto_delete);
                    req.mutable_args()->swap(args);

                    _codecP->send(_connectionP,req);
                    basicCommonResponsePtr resp = waitresponse(rid);
                    return resp->ok();
                }
            void deleteQueue(const std::string& queue_name){
                    std::string rid = Uuidhelper::BuideUuid();
                    deleteQueueRequest req;
                    req.set_rid(rid);
                    req.set_cid(_cid);
                    req.set_queue_name(queue_name);

                    _codecP->send(_connectionP,req);
                    waitresponse(rid);
                    return ;
            }
            bool bind (const std::string& exchange_name,
                const std::string& queue_name,
                const std::string& binding_key){
                    std::string rid = Uuidhelper::BuideUuid();
                    queueBindRequest req;
                    req.set_rid(rid);
                    req.set_cid(_cid);
                    req.set_queue_name(queue_name);
                    req.set_exchange_name(exchange_name);
                    req.set_binding_key(binding_key);

                    _codecP->send(_connectionP,req);
                    basicCommonResponsePtr resp = waitresponse(rid);
                    return resp->ok();
                }
            void unBind(const std::string& exchange_name,const std::string& queue_name){
                    std::string rid = Uuidhelper::BuideUuid();
                    queueUnBindRequest req;
                    req.set_rid(rid);
                    req.set_cid(_cid);
                    req.set_queue_name(queue_name);
                    req.set_exchange_name(exchange_name);

                    _codecP->send(_connectionP,req);
                    waitresponse(rid);
                    return ;
            }
            bool basicPublish(const std::string& exchange_name,const std::string& body,BasicProperty* property){
                    std::string rid = Uuidhelper::BuideUuid();
                    basicPublishRequest req;
                    req.set_rid(rid);
                    req.set_cid(_cid);
                    req.set_exchange_name(exchange_name);
                    req.set_body(body);
                    if(property != nullptr){
                        req.mutable_property()->set_id(property->id());
                        req.mutable_property()->set_deliverymod(property->deliverymod());
                        req.mutable_property()->set_routingkey(property->routingkey());
                    }
                    _codecP->send(_connectionP,req);
                    basicCommonResponsePtr resp = waitresponse(rid);
                    return resp->ok();
                    
            }
            void basicAck(const std::string& message_id){
                    std::string rid = Uuidhelper::BuideUuid();
                    basicAckRequest req;
                    req.set_rid(rid);
                    req.set_cid(_cid);
                    req.set_queue_name(_consumer->qname);
                    req.set_message_id(message_id);

                    _codecP->send(_connectionP,req);
                    waitresponse(rid);
                    return ;
            }
            bool basicConsume(const std::string& consumer_tag,const std::string& queue_name,bool auto_ack,
                const ConsumerCallback& cb){
                if(_consumer.get() != nullptr){
                    LOG_ERRORM("此消费者已订阅...");
                    return false;
                }
                std::string rid = Uuidhelper::BuideUuid();
                basicConsumerRequest req;
                req.set_rid(rid);
                req.set_cid(_cid);
                req.set_queue_name(queue_name);
                req.set_auto_ack(auto_ack);
                req.set_consumer_tag(consumer_tag);
                _codecP->send(_connectionP,req);

                basicCommonResponsePtr resp = waitresponse(rid);

                if(!resp->ok()) {
                    LOG_ERRORM("订阅失败...");
                    return false;
                }
                _consumer = std::make_shared<Consumer>(consumer_tag,queue_name,auto_ack,cb);
                LOG_DEBUGM("创建消费者 done...");
                return true;
            }
            void basicCancel(){
                if(_consumer.get() == nullptr){
                    LOG_ERRORM("无消费者...");
                    return ;
                }
                std::string rid = Uuidhelper::BuideUuid();
                basicCancelRequest req;
                req.set_rid(rid);
                req.set_cid(_cid);
                req.set_queue_name(_consumer->qname);
                req.set_consumer_tag(_consumer->tag);

                _codecP->send(_connectionP,req);
                waitresponse(rid);
                _consumer.reset();
                return ;
            }
        public:
            void putBasicResponse(const basicCommonResponsePtr& resp){
                std::unique_lock<std::mutex> lock(_mutex);
                _basic_rep.insert({resp->rid(),resp});
                _cv.notify_all();
            }
            
            void consume(const basicConsumerResponsePtr& resp){
                if(resp.get() == nullptr){
                    LOG_ERRORM("basicConsumerResponse 为空...");
                    return ;
                }
                if(_consumer.get() == nullptr){
                    LOG_ERRORM("无消费者...");
                    return ;
                }
                if(_consumer->tag != resp->consumer_tag()){
                    LOG_ERRORM("消费者不一样错误...");
                    return ;
                }
                _consumer->cb(resp->consumer_tag(),resp->mutable_property(),resp->body());

            }

        private:    
            basicCommonResponsePtr waitresponse(const std::string& rid)
            {
                std::unique_lock<std::mutex> lock(_mutex);
                _cv.wait(lock,[this,&rid](){
                    return _basic_rep.find(rid) != _basic_rep.end();
                });
                auto it = _basic_rep.find(rid);
                if (it == _basic_rep.end()) {
                    LOG_ERROR << "Response not found for rid: " << rid << " after wait";
                    return nullptr;
                }
                basicCommonResponsePtr resp = it->second;
                _basic_rep.erase(it);
                return resp;
            }

        private:
            std::string _cid;
            Consumer::ptr _consumer;
            muduo::net::TcpConnectionPtr _connectionP;
            ProtobufCodecPtr _codecP;
            std::mutex _mutex;
            std::condition_variable _cv;
            std::unordered_map<std::string,basicCommonResponsePtr> _basic_rep;
    };

    class ClientChannelManager{
        public:
            using ptr = std::shared_ptr<ClientChannelManager>;
            
            ClientChannelManager(){}

            ClientChannel::ptr create(const muduo::net::TcpConnectionPtr& conn,const ProtobufCodecPtr& codecp){
                std::unique_lock<std::mutex> lock(_mutex);
                ClientChannel::ptr cchannel = std::make_shared<ClientChannel>(conn,codecp);
                _channels.insert(std::make_pair(cchannel->getCid(),cchannel));
                return cchannel;
            }
            void remove(const std::string& cid){
                std::unique_lock<std::mutex> lock(_mutex);
                auto it = _channels.find(cid);
                if(it == _channels.end()){
                    return;
                }
                _channels.erase(it);
            }
            ClientChannel::ptr getChannel(const std::string& cid){
                auto it = _channels.find(cid);
                if(it == _channels.end()){
                    return ClientChannel::ptr();
                }
                return it->second;
            }
            
        private:
            std::mutex _mutex;
            std::unordered_map<std::string,ClientChannel::ptr> _channels;
    };

}

#endif